package com.sfx.concurrent.demo.concurrent.BlockingQueueTest;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class MyBlockingQueueDoubleCheck<E> {
    private final E[] queue;
    private int capacity;
    private AtomicInteger size = new AtomicInteger(0);
    private int head;
    private int tail;

    public MyBlockingQueueDoubleCheck(int capacity) {
        this.capacity = capacity;
        queue = (E[]) new Object[capacity];
    }
    private final ReentrantLock headLock = new ReentrantLock();
    private final ReentrantLock tailLock = new ReentrantLock();

    private final Condition headCondition = headLock.newCondition();
    private final Condition tailCondition = tailLock.newCondition();

    public void offer(E e) throws InterruptedException {
        tailLock.lockInterruptibly();
        int c;
        try{
            while(isFull()) {
                tailCondition.await();// offer1 offer2 offer3
            }
            queue[tail] = e;
            if(++tail == this.capacity) {
                tail = 0;
            }
            //size++
            c = size.getAndIncrement();
            //放进元素后仍然有空位,那么由offer线程级联唤醒
            if(c+1 < queue.length) {
                tailCondition.signal();
            }
        }finally {
            tailLock.unlock();
        }
        //唤醒因队列为空等待的线程,注意await/signal 都要配合lock一起使用-->防止死锁所以放在外面
        //我们要保证唤醒时加锁次数降低到最低->只要唤醒headCondition一次,剩下的都由它自己去唤醒
        //当来了多个offer线程的时候,让一个offer线程去唤醒pool线程降低加锁次数,剩下的都由pool线程自己级联唤醒
        if(c == 0) {
            headLock.lock();// offer1 offer2 offer3
            try{
                //不满了可以取出元素了
                headCondition.signal();
            }finally {
                headLock.unlock();
            }
        }
    }

    public boolean offer(E e,long timeout) throws InterruptedException {
        long nanos = TimeUnit.MILLISECONDS.toNanos(timeout);
        tailLock.lockInterruptibly();
        try {
            while(isEmpty()) {
                if(nanos <=0) {
                    return false;
                }
                nanos = tailCondition.awaitNanos(nanos);
            }
            queue[tail] = e;
            if(++tail == this.capacity) {
                tail = 0;
            }
            size.getAndIncrement();
        }finally {
            tailLock.unlock();
        }

        headLock.lockInterruptibly();
        try{
            tailCondition.signal();
        }finally {
            headLock.unlock();
        }
        return true;
    }

    public E pool() throws InterruptedException {
        E e;
        int c;
        headLock.lockInterruptibly();
        try{
            while(isEmpty()) {//pool_1 pool_2 pool_3
                headCondition.await();
            }
            e = queue[head];
            queue[head] = null;
            if(++head == this.capacity) {
                head = 0;
            }
            //size--
            c = size.getAndDecrement();
            //当来了多个offer线程的时候,让一个offer线程去唤醒pool线程降低加锁次数,剩下的都由pool线程自己级联唤醒
            if(c>1) {
                headCondition.signal();
            }
        }finally {
            headLock.unlock();//pool1 pool2 pool3
        }

        //唤醒因队列满而等待的线程
        //当有多个pool线程来弹出元素时,只让一个pool线程加锁去唤醒一个offer线程,剩下的要让offer线程自己级联唤醒
        //队列由满->不满 唤醒一个offer线程,剩下的都由offer线程自己级联唤醒
        if(c == queue.length) {
            tailLock.lock();
            try {
                tailCondition.signal();
            } finally {
                tailLock.unlock();
            }
        }
        return e;
    }

    private boolean isEmpty() {
        return this.size.get() == 0;
    }

    private boolean isFull() {
        return this.size.get()==this.capacity;
    }
}